package p.ithorns.integration.hazelcast.cdc;

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.jet.JetService;
import com.hazelcast.jet.cdc.CdcSinks;
import com.hazelcast.jet.cdc.ChangeRecord;
import com.hazelcast.jet.cdc.mysql.MySqlCdcSources;
import com.hazelcast.jet.config.JobConfig;
import com.hazelcast.jet.pipeline.Pipeline;
import com.hazelcast.jet.pipeline.StreamSource;
import p.ithorns.integration.db.model.Customer;

/**
 * Producer
 * TODO ithorns
 *
 * @author Ht.L
 * @date 2023-10-18 08:48
 * @since 1.0.0
 */
public class Producer {

    private static void bootCdc() {
        StreamSource<ChangeRecord> source = MySqlCdcSources.mysql("source")
                .setDatabaseAddress("127.0.0.1")
                .setDatabasePort(3306)
                .setDatabaseUser("root")
                .setDatabasePassword("Z@9e59b$")
                .setClusterName("local")
                // .setDatabaseWhitelist("hazelcast")
                .setTableWhitelist("hazelcast.customers")
                .setCustomProperty("database.connectionTimeZone", "Asia/Shanghai")
                .setCustomProperty("database.allowPublicKeyRetrieval", "true")
                .build();

        Pipeline pipeline = Pipeline.create();
        pipeline.readFrom(source)
                .withoutTimestamps()
                .peek()
                .writeTo(CdcSinks.map("customers",
                        r -> {
                            return r.key().toMap().get("id");
                        },
                        r -> {
                            return r.value().toObject(Customer.class).toString();
                        }));

        JobConfig cfg = new JobConfig().setName("mysql-monitor");
        HazelcastInstance hz = Hazelcast.bootstrappedInstance();
        JetService jet = hz.getJet();
        jet  .newJob(pipeline, cfg);
    }

}